package cn.wolfcode.service.impl;

import cn.wolfcode.common.exception.BusinessException;
import cn.wolfcode.common.web.CodeMsg;
import cn.wolfcode.common.web.Result;
import cn.wolfcode.domain.Product;
import cn.wolfcode.domain.SeckillProduct;
import cn.wolfcode.domain.SeckillProductVo;
import cn.wolfcode.feign.ProductFeignApi;
import cn.wolfcode.mapper.SeckillProductMapper;
import cn.wolfcode.redis.SeckillRedisKey;
import cn.wolfcode.service.ISeckillProductService;
import cn.wolfcode.common.utils.AssertUtil;
import cn.wolfcode.web.msg.SeckillCodeMsg;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.cache.annotation.CacheConfig;
import org.springframework.cache.annotation.CacheEvict;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
@Service
@CacheConfig(cacheNames = "SeckillProduct")
public class SeckillProductServiceImpl implements ISeckillProductService {
    private final SeckillProductMapper seckillProductMapper;
    private final StringRedisTemplate redisTemplate;
    private final ProductFeignApi productFeignApi;
    private final DistributedLockSupport distributedLockSupport;

    public SeckillProductServiceImpl(SeckillProductMapper seckillProductMapper, StringRedisTemplate redisTemplate, ProductFeignApi productFeignApi, DistributedLockSupport distributedLockSupport) {
        this.seckillProductMapper = seckillProductMapper;
        this.redisTemplate = redisTemplate;
        this.productFeignApi = productFeignApi;
        this.distributedLockSupport = distributedLockSupport;
    }

    @Override
    public List<SeckillProductVo> selectTodayListByTime(Integer time) {
        // 1. 调用秒杀服务接口, 基于今天的时间, 查询今天的所有秒杀商品数据
        List<SeckillProduct> todayList = seckillProductMapper.queryCurrentlySeckillProduct(time);
        // 2. 遍历秒杀商品列表, 得到商品 id 列表
        List<Long> productIdList = todayList.stream() // Stream<SeckillProduct>
                .map(SeckillProduct::getProductId) // SeckillProduct => Long
                .distinct()
                .collect(Collectors.toList());
        // 3. 根据商品 id 列表, 调用商品服务查询接口, 得到商品列表
        Result<List<Product>> result = productFeignApi.selectByIdList(productIdList);
        /**
         * result 可能存在的几种情况:
         *  1. 远程接口正常返回, code == 200, data == 想要的数据
         *  2. 远程接口出现异常, code != 200
         *  3. 接口被熔断降级, data == null
         */
        if (result.hasError() || result.getData() == null) {
            throw new BusinessException(new CodeMsg(result.getCode(), result.getMsg()));
        }

        List<Product> products = result.getData();

        // 4. 遍历秒杀商品列表, 将商品对象与秒杀商品对象聚合到一起
        // List<SeckillProduct> => List<SeckillProductVo>
        List<SeckillProductVo> productVoList = todayList.stream()
                .map(sp -> {
                    SeckillProductVo vo = new SeckillProductVo();
                    BeanUtils.copyProperties(sp, vo);

                    List<Product> list = products.stream().filter(p -> sp.getProductId().equals(p.getId())).collect(Collectors.toList());
                    if (list.size() > 0) {
                        Product product = list.get(0);
                        BeanUtils.copyProperties(product, vo);
                    }
                    vo.setId(sp.getId());

                    return vo;
                }) // Stream<SeckillProductVo>
                .collect(Collectors.toList());

        return productVoList;
    }

    @Override
    public List<SeckillProductVo> selectTodayListByTimeFromRedis(Integer time) {
        String key = SeckillRedisKey.SECKILL_PRODUCT_LIST.join(time + "");
        List<String> stringList = redisTemplate.opsForList().range(key, 0, -1);

        if (stringList == null || stringList.size() == 0) {
            log.warn("[秒杀商品] 查询秒杀商品列表异常, Redis 中没有数据, 从 DB 中查询...");
            return this.selectTodayListByTime(time);
        }

        return stringList.stream().map(json -> JSON.parseObject(json, SeckillProductVo.class)).collect(Collectors.toList());
    }

    @Override
    @Cacheable(key = "'selectByIdAndTime:' + #time + ':' + #seckillId")
    public SeckillProductVo selectByIdAndTime(Long seckillId, Integer time) {
        SeckillProduct seckillProduct = seckillProductMapper.selectByIdAndTime(seckillId, time);

        Result<List<Product>> result = productFeignApi.selectByIdList(Collections.singletonList(seckillProduct.getProductId()));
        if (result.hasError() || result.getData() == null || result.getData().size() == 0) {
            throw new BusinessException(new CodeMsg(result.getCode(), result.getMsg()));
        }
        Product product = result.getData().get(0);

        SeckillProductVo vo = new SeckillProductVo();
        // 先将商品的属性 copy 到 vo 对象中
        BeanUtils.copyProperties(product, vo);

        // 再将秒杀商品的属性 copy 到 vo 对象中, 并覆盖 id 属性
        BeanUtils.copyProperties(seckillProduct, vo);
        return vo;
    }

    /**
     * 分布式总结：
     * 1.通过 Redis 的 SETNX 命令实现分布式锁
     * 2.为避免死锁，加锁时引入超时机制，如果出现意外宕机，可以实现自动释放锁
     * 3.为避免加锁与设置超时时间命令执行的原子性问题，通过lua脚本保证多个命令执行的原子性
     * 4.锁只能由加锁的线程释放，不能被其他线程释放，利用唯一线程id作为value,删除时判断id是否一致
     * 5.为避免业务没执行完成，而锁就过期，引入 WatchDog 看门狗机制实现锁自动续期
     */
    /*@CacheEvict(key = "'selectByIdAndTime:' + #time + ':' + #id")
    @Override
    public void decrStockCount(Long id, Integer time) {
        //自定义分布式锁思路
        //1.锁哪个对象 =》 锁指定场次下的指定商品
        //2.当多线程同时加锁，只能有一个线程加锁成功 =》 Redis的setnx命令(key存在则不做，key不存在则保存)，如：使用 setnx 秒杀场次+秒杀商品id xxx
        //3.锁记录在什么地方 => 利用Redis的 setnx 命令存储在Redis的String数据结构中
        String lockKey = "seckill:product:stockcount:" + time + ":" + id;
        //雪花算法：生成分布式唯一id，作为线程的id,来保证自己释放自己的分布式锁
        String threadId = IdGenerateUtil.get().nextId() + "";
        ScheduledFuture<?> future = null;
        try {
            int count = 0;
            int timeout = 5;
            //4.当线程获取不到锁的时候，执行什么策略 => 阻塞/自旋等待(限制次数)/直接抛异常
            //如果自旋次数超过5次，抛出异常
            do {
                //加锁
//                Boolean ret = redisTemplate.opsForValue().setIfAbsent(key, "1");
                //避免死锁的问题：使用Lua脚本，一次性发送 setnx 和 设置过期时间的命令给redis 执行
                Boolean ret = redisTemplate.execute(lockScript, Collections.singletonList(lockKey), threadId, timeout + "");
                if (ret != null && ret) {
                    break;
                }
                AssertUtil.isTrue((count++) < 5, "系统繁忙,请稍后再试!");
                //避免CPU过于频繁
                TimeUnit.MILLISECONDS.sleep(20);
            } while (true);

            //加锁成功后，创建 WatchDog 看门狗监听 业务是否执行完成(Redis中的key是否存在)，实现续期操作
            long delayTime = (long) (timeout * 0.8);
            System.out.println("[WatchDog]--------------- 间隔时间："+delayTime);
            future = scheduledThreadPoolExecutor.scheduleAtFixedRate(
                    () -> {
                        //从redis查询key是否存在，如果存在，则续期
                        String value = redisTemplate.opsForValue().get(lockKey);
                        if (StringUtils.isNotEmpty(value)) {
                            redisTemplate.expire(lockKey, delayTime + 2, TimeUnit.SECONDS);
                            System.out.println("[WatchDog] --------------------执行Redis实现续期操作 threadId=" + threadId + " key=" + lockKey + " value=" + value);
                        }
                    },
                    delayTime,
                    delayTime,
                    TimeUnit.SECONDS
            );

            //模拟业务11s,看续期效果
//            TimeUnit.SECONDS.sleep(11);

            //业务逻辑代码-----------------------------开始
            //1.查询MySQL中秒杀库存数量
            int stockCount = seckillProductMapper.getStockCount(id);
            AssertUtil.isTrue(stockCount > 0, "库存不足");
            //2.扣除库存
            seckillProductMapper.decrStock(id);
            //业务逻辑代码------------------------------结束

        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            //业务执行完成，则销毁key续期操作任务，取消WatchDog监听
            if (future != null) {
                //如果该任务还在执行，是否停止。如果走到finally代表业务逻辑已经执行完成，因此不需要该任务去执行续期操作了
                future.cancel(true);
            }

            //释放锁
            //在释放分布式锁之前，获取该分布式锁，并判断该分布式锁是否是当前线程的分布式锁，是才可以释放。自己释放自己的分布式锁
            //如果该key在未执行完业务代码则自动过期，可能获取到null,因此判断要通过不为null值进行equals,否则出现空指针异常
//            String value = redisTemplate.opsForValue().get(key);
//            if (threadId.equals(value)) {
//                redisTemplate.delete(key);
//            }
            //把上述释放锁的代码优化到lua脚本中执行
            redisTemplate.execute(unlockScript,Collections.singletonList(lockKey),threadId);
        }
    }*/
    @CacheEvict(key = "'selectByIdAndTime:' + #time + ':' + #id")
    @Override
    public void decrStockCount(Long id, Integer time) {
        //Redis实现分布式锁方式解决库存超卖问题
        /*String lockKey = CommonRedisKey.SECKILL_STOCK_COUNT_LOCK_KEY_STRING.getRealKey(time, id);
        int timeout = 5;
        distributedLockSupport.lock(lockKey, timeout, () -> {
            //业务逻辑代码-----------------------------开始
            //1.查询MySQL中秒杀库存数量
            int stockCount = seckillProductMapper.getStockCount(id);
            AssertUtil.isTrue(stockCount > 0, SeckillCodeMsg.SECKILL_STOCK_OVER.getMsg());
            //2.扣除库存
            seckillProductMapper.decrStock(id);
            //业务逻辑代码------------------------------结束
        });*/

        //乐观锁方式解决库存超卖
        int row = seckillProductMapper.decrStock(id);
        AssertUtil.isTrue(row > 0,SeckillCodeMsg.SECKILL_STOCK_OVER.getMsg());
    }

    @Override
    public int selectStockCountById(Long seckillId) {
        return seckillProductMapper.getStockCount(seckillId);
    }
}
